package com.atguigu.userprofile.app;

import com.atguigu.userprofile.common.bean.TagInfo;
import com.atguigu.userprofile.common.bean.TaskInfo;
import com.atguigu.userprofile.common.bean.TaskTagRule;
import com.atguigu.userprofile.common.constants.CodeConst;
import com.atguigu.userprofile.common.dao.TagInfoDAO;
import com.atguigu.userprofile.common.dao.TaskInfoDAO;
import com.atguigu.userprofile.common.dao.TaskTagRuleDAO;
import com.atguigu.userprofile.common.util.MyPropertiesUtil;
import com.atguigu.userprofile.common.util.MySQLUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.SparkSession;

import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class TaskSQLApp {

//1      获得标签定义
//       从mysql中读取3张表  ： task_info  tag_info  task_tag_rule
//          jdbc

    public static void main(String[] args) {
        //0  接受spark-submit 结尾的业务参数  2个：  1.  task_id  2 .  busi_date
        String taskId = args[0];
        String busiDate = args[1];
        // spark 环境
        SparkConf sparkConf = new SparkConf().setAppName("task_sql_app");//.setMaster("local[*]");
        SparkSession sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate();

//1      获得标签定义
//       从mysql中读取3张表  ：  tag_info  task_info  task_tag_rule
//          jdbc


        TagInfo tagInfo = TagInfoDAO.getTagInfoByTaskId(taskId);
      //完成：
        TaskInfo taskInfo= TaskInfoDAO.getTaskInfoById(taskId);

        List<TaskTagRule>  taskTagRuleList= TaskTagRuleDAO.getTaskTagRuleList(taskId);


        System.out.println(tagInfo);
        System.out.println(taskInfo);
        System.out.println(taskTagRuleList);



        //2 自动建表  每个标签一张表
        //Create table if not exists  $tableName
        //( uid string ,   tag_value   fieldType   )
        //Partitioned by (dt  string )
        //格式？ 压缩？ 不压缩  文本 \t 分隔 ROW FORMAT DELIMITED FIELDS TERMINATED BY '\\t'
        //Location   hdfs://xxx:8020/xxx/库名/表名

        String tableName=tagInfo.getTagCode().toLowerCase();

        String fieldType="";
        if(tagInfo.getTagValueType().equals(CodeConst.TAG_VALUE_TYPE_LONG)){
            fieldType="bigint";
        } else if (tagInfo.getTagValueType().equals(CodeConst.TAG_VALUE_TYPE_DECIMAL)){
            fieldType = "decimal(16,2)";
        } else if (tagInfo.getTagValueType().equals(CodeConst.TAG_VALUE_TYPE_STRING)){
            fieldType = "string";
       } else if (tagInfo.getTagValueType().equals(CodeConst.TAG_VALUE_TYPE_DATE)){
           fieldType = "string";
       }

        Properties properties = MyPropertiesUtil.load("config.properties");
        String hdfsPath = properties.getProperty("hdfs-store.path");
        String dwName = properties.getProperty("data-warehouse.dbname");
        String upName = properties.getProperty("user-profile.dbname");
        String createTableSQL="     create table if not exists   "+upName+"." +tableName+
                "      ( uid string ,   tag_value   "+fieldType+"   )\n" +
                "       partitioned by (dt  string )\n" +
                "        ROW FORMAT DELIMITED FIELDS TERMINATED BY '\\t'\n" +
                "        location   '"+hdfsPath+"/"+upName+"/"+tableName+"'";
        System.out.println(createTableSQL);

        sparkSession.sql(createTableSQL);

        //3 拼接select sql
        // select uid ,case query_value when 'M' then '男' when 'F' then '女‘ when ... end  as tag_value
        //(  $tagsql ) tt
          String tagValueSQL="";
          //作业的兼容性问题1：  如果没有四级标签的定义，直接用查询结果作为标签值
        if(taskTagRuleList.size()>0 ){
            List<String> whenThenList = taskTagRuleList.stream().map(taskTagRule -> "when '" + taskTagRule.getQueryValue() + "' then '" + taskTagRule.getSubTagValue()+"'").collect(Collectors.toList());
            String whenThenSQL = StringUtils.join(whenThenList, " ");
            String caseWhenSQL="case query_value "+ whenThenSQL +" end  as tag_value";
            tagValueSQL=caseWhenSQL;
        }else{
            tagValueSQL=" query_value as tag_value";
        }

        //作业兼容性问题2  ：  如果由时间指代符，则替换为当前业务时间
        String taskSQL = taskInfo.getTaskSql().replace("$dt", busiDate);

        String selectSQL="select uid , "+tagValueSQL+" from ("+ taskSQL+") tt";

        System.out.println(selectSQL);

        // 4 insert sql
        String insertSQL="insert overwrite table "+upName+"."+tableName+" partition (dt= '"+busiDate+"') "+selectSQL;
        System.out.println(insertSQL);
        sparkSession.sql("use "+dwName);
        sparkSession.sql(insertSQL);



    }
}
